package com.wsjj.gmall.util;



import com.alibaba.fastjson.JSONObject;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Set;

// Hbase连接工具类
public class HbaseUtil {
    private static final Logger log = LoggerFactory.getLogger(HbaseUtil.class);


//TODO    1.连接Hbase连接

    /**
     *
     * @return
     * @throws IOException
     * 返回Connection连接器
     */
    public static Connection getHBseConnection() throws IOException {
        Configuration conf = new Configuration();
        conf.set("hbase.zookeeper.quorum", "hadoop102,hadoop103,hadoop101");
        conf.set("hbase.zookeeper.property.clientPort", "2181");

        return ConnectionFactory.createConnection(conf);

    }

//TODO    2.关闭Hbase

    /**
     *
     * @param hbaseConn 传入连接器
     */
    public static void closeHBaseConn(Connection hbaseConn){
        if(hbaseConn != null && !hbaseConn.isClosed()){
            try {
                hbaseConn.close();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }

//TODO    3.创建表

    /**
     *
     * @param hbaseConn 连接器
     * @param nameSpace 表空间(相当于mysql的库)
     * @param table     表名
     * @param families  列族
     */
    public static void createHBaseTable(Connection hbaseConn,String nameSpace,String table,String ... families){
        if(families.length < 1){
            System.out.println("至少需要一个列族");
        }

        try (Admin admin = hbaseConn.getAdmin()){

            TableName tablenameObj = TableName.valueOf(nameSpace, table);
            if (admin.tableExists(tablenameObj)){
                System.out.println("表空间"+nameSpace+"下的表"+table+"已经存在");
                return;
            }

//            添加    列族
            TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tablenameObj);
            for (String family : families) {
                ColumnFamilyDescriptor columnFamilyDescriptor = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(family)).build();
                tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptor);
            }

//            创建表
            admin.createTable(tableDescriptorBuilder.build());
            System.out.println("表空间"+nameSpace+"下的表"+table+"创建成功");


        } catch (IOException e) {
            throw new RuntimeException(e);
        }


    }

//TODO    4.删除表

    /**
     *
     * @param hbaseConn  连接器
     * @param nameSpace     表空间(相当于mysql的库)
     * @param table         表名
     */
    public static void dropHBaseTable(Connection hbaseConn,String nameSpace,String table) throws IOException {
        Admin admin = hbaseConn.getAdmin();

        TableName tableName = TableName.valueOf(nameSpace, table);
        if(admin.tableExists(tableName)){
            admin.disableTable(tableName);
            admin.deleteTable(tableName);
        }
        admin.close();
        log.info(nameSpace+" "+table+" "+"删除成功");

    }

//TODO  5.插入数据

    /**
     *
     * @param hbaseConn  连接器
     * @param nameSpace 表空间(相当于mysql的库)
     * @param table     表
     * @param rowkey    rowkey
     * @param family    列族
     * @param jsonObj   插入的数据
     */
    public static void putRow(Connection hbaseConn, String nameSpace, String table, String rowkey, String family, JSONObject jsonObj){
        TableName tableName = TableName.valueOf(nameSpace, table);
        try(Table table1 = hbaseConn.getTable(tableName)) {

            Put put = new Put(Bytes.toBytes(rowkey));
            Set<String> keys = jsonObj.keySet();
            for (String key : keys) {
                String value = jsonObj.getString(key);
                if(StringUtils.isNotEmpty(value)){
                    put.addColumn(Bytes.toBytes(family), Bytes.toBytes(key),Bytes.toBytes(value));
                }
            }

            table1.put(put);
            System.out.println("表空间"+nameSpace+"下的表"+table+"中put数据成功");

        } catch (IOException e) {
            throw new RuntimeException(e);
        }


    }




//TODO  6.删除数据

    public static void delRow(Connection hbaseConn, String nameSpace, String table, String rowkey){

        TableName tableName = TableName.valueOf(nameSpace,table);
        try(Table connTable = hbaseConn.getTable(tableName)) {
            Delete delete = new Delete(Bytes.toBytes(rowkey));
            connTable.delete(delete);

        } catch (IOException e) {
            throw new RuntimeException(e);
        }

    }

}
